package com.xiaohu.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * join算子也要作用在kv格式的RDD上
 */
object Demo9Join {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("reduceByKey")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    //parallelize：将scala的集合变成spark中的RDD
    val rdd1: RDD[(String, String)] = sc.parallelize(List(
      ("1001", "易政"),
      ("1002", "盛宇豪"),
      ("1003", "雷彭程"),
      ("1004", "周钰"),
      ("1005", "彭飞")
    ))

    val rdd2: RDD[(String, String)] = sc.parallelize(List(
      ("1001", "看美女"),
      ("1002", "看综艺"),
      ("1003", "看八卦"),
      ("1004", "打游戏"),
      ("1009", "学习")
    ))

    /**
     * join 内连接
     * right join 右连接
     * left join 左连接
     * full join 全连接
     */
    // join 内连接 两个rdd共同拥有的键才会进行关联
    //    val resRDD1: RDD[(String, (String, String))] = rdd1.join(rdd2)
    //    val resRDD2: RDD[(String, String, String)] = resRDD1.map {
    //      case (id: String, (name: String, like: String)) =>
    //        (id, name, like)
    //    }
    //    resRDD2.foreach(println)

    //right join 右连接 保证右边rdd键的完整性
    //    val resRDD2: RDD[(String, (Option[String], String))] = rdd1.rightOuterJoin(rdd2)
    //    val resRDD3: RDD[(String, String, String)] = resRDD2.map {
    //      case (id: String, (Some(name), like: String)) =>
    //        (id, name, like)
    //      case (id: String, (None, like: String)) =>
    //        (id, "查无此人", like)
    //    }
    //    resRDD3.foreach(println)

    //TODO:自己完成左关联

    //全关联
    val resRDD2: RDD[(String, (Option[String], Option[String]))] = rdd1.fullOuterJoin(rdd2)
    val resRDD3: RDD[(String, String, String)] = resRDD2.map {
      case (id: String, (Some(name), Some(like))) =>
        (id, name, like)
      case (id: String, (Some(name), None)) =>
        (id, name, "此人无爱好")
      case (id: String, (None, Some(like))) =>
        (id, "查无此人", like)
    }
    resRDD3.foreach(println)


  }
}
